package otSpaces.server;

import java.io.Serializable;
import java.math.BigInteger;
import java.nio.ByteBuffer;
import java.util.HashSet;
import java.util.Iterator;
import java.util.NoSuchElementException;
import java.util.Set;
import java.util.logging.Logger;

import otSpaces.server.util.AmbiguousReference;

import com.sun.sgs.app.Channel;
import com.sun.sgs.app.ChannelListener;
import com.sun.sgs.app.ClientSession;
import com.sun.sgs.app.Delivery;
import com.sun.sgs.app.ManagedObjectRemoval;

/**
 * Implements a single channel, within a single topicSpace. Users should not
 * need to access this class directly as it will masquerade as a Channel
 * 
 * @author peter.lavalle@gmail.com
 * 
 */
class OtSpaceTopic implements Channel, Serializable, ManagedObjectRemoval {

	public static final Logger LOG = Logger.getLogger(OtSpaceTopic.class
			.getName());

	private static final long serialVersionUID = -4544212274604440395L;
	private final HashSet<BigInteger> _clientIds = new HashSet<BigInteger>();
	public final AmbiguousReference<ChannelListener> _listener;
	private final AmbiguousReference<OtSpaceManager> _managerRef;

	public final String _name;

	OtSpaceTopic(ChannelListener listener, final OtSpaceManager manager,
			final String name) {
		_listener = new AmbiguousReference<ChannelListener>(listener);
		_name = name;
		_managerRef = new AmbiguousReference<OtSpaceManager>(manager);
	}

	@Override
	public Delivery getDelivery() {
		return Delivery.RELIABLE;
	}

	@Override
	public String getName() {
		return _name;
	}

	@Override
	public Iterator<ClientSession> getSessions() {

		final Iterator<BigInteger> ids = _clientIds.iterator();

		return new Iterator<ClientSession>() {

			ClientSession _next = null;

			@Override
			public boolean hasNext() {
				if (_next == null) {

					while (ids.hasNext() && _next == null) {
						final BigInteger nextId = ids.next();

						_next = _managerRef.get().clientSession(nextId);
					}
				}

				return _next != null;
			}

			@Override
			public ClientSession next() {
				hasNext();
				if (_next == null) {
					throw new NoSuchElementException();
				} else {
					final ClientSession ret = _next;
					_next = null;
					return ret;
				}
			}

			@Override
			public void remove() {
				if (_next == null) {
					throw new IllegalStateException();
				} else {
					ids.remove();
				}
			}

		};

	}

	@Override
	public boolean hasSessions() {
		return getSessions().hasNext();
	}

	public Channel join(BigInteger sessionId) {

		if (!_clientIds.contains(sessionId)) {

			// send the subscriber a join message
			if (_managerRef.get().clientJoin(sessionId, getName())) {

				// add the id to us
				_clientIds.add(sessionId);
			} else {
				LOG.warning("[" + _managerRef.get().getName() + ">" + getName()
						+ "] A client has gone missing ");
			}
		}
		return this;
	}

	@Override
	public Channel join(ClientSession session) {

		// get the id from the manager
		final BigInteger id = _managerRef.get().clientId(session);

		// call the 'other' method
		return join(id);
	}

	public Channel join(Iterator<? extends ClientSession> sessions) {

		while (sessions.hasNext()) {
			join(sessions.next());
		}

		return this;
	}

	@Override
	public Channel join(Set<? extends ClientSession> sessions) {
		return join(sessions.iterator());
	}

	public Channel leave(BigInteger sessionId) {

		if (_clientIds.contains(sessionId)) {
			// send the subscriber a join message
			if (!_managerRef.get().clientLeave(sessionId, getName())) {
				LOG.warning("[" + _managerRef.get().getName() + ">" + getName()
						+ "] A client has gone missing ");
			}

			_clientIds.remove(sessionId);
		}

		return this;
	}

	@Override
	public Channel leave(ClientSession session) {

		// get the id from the manager
		final BigInteger id = _managerRef.get().clientId(session);

		// call the 'other' method
		return leave(id);
	}

	public Channel leave(Iterator<? extends ClientSession> sessions) {

		while (sessions.hasNext()) {
			leave(sessions.next());
		}

		return this;
	}

	@Override
	public Channel leave(Set<? extends ClientSession> sessions) {
		return leave(sessions.iterator());
	}

	@Override
	public Channel leaveAll() {
		return leave(getSessions());
	}

	@Override
	public void removingObject() {

		leaveAll();

		// assert that our subscriber thing is empty
		assert _clientIds.isEmpty();
	}

	@Override
	public Channel send(ByteBuffer message) {

		// grab the manager
		final OtSpaceManager manager = _managerRef.get();

		// for each subscriber
		for (BigInteger id : new HashSet<BigInteger>(_clientIds)) {

			if (!manager.clientSend(id, getName(), message)) {
				LOG.warning("[" + _managerRef.get().getName() + ">" + getName()
						+ "] A client has gone missing ");
				_clientIds.remove(id);
			}
		}

		return this;
	}

	@Override
	public Channel send(ClientSession sender, ByteBuffer message) {
		throw new UnsupportedOperationException("Doesn't support sender");
	}

}
